package org.redis.cluster.loadbalance.slot;

import java.util.Collection;

import org.apache.commons.lang3.StringUtils;
import org.redis.cluster.RedisCluster;
import org.redis.cluster.config.ClientConfig;
import org.redis.cluster.loadbalance.HashLoadBalance;
import org.redis.cluster.plugin.Plugin;
import org.redis.cluster.redis.RedisCommand;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SlotLoadBalance extends HashLoadBalance implements Plugin {

	protected final Logger logger=LoggerFactory.getLogger(getClass());
	
	private RedisCluster redisCluster;
	private SlotManager slotManager;
	private SlotReplicater slotReplicater;
	
	@Override    
	public void init(RedisCluster redisCluster) {
		this.redisCluster=redisCluster;
		this.slotManager=initSlotManager();
		this.slotReplicater=initSlotReplicater();
	}
	
	protected SlotManager initSlotManager() {
		SlotManager slotManager=new SlotManager();
		slotManager.init(redisCluster);
		return slotManager;
	}
	
	protected SlotReplicater initSlotReplicater() {
		SlotReplicater slotReplicater=new SlotReplicater(slotManager);
		slotReplicater.init(redisCluster);
		return slotReplicater;
	}
	
	@Override
	public void destroy() {
		slotManager.destroy();
		slotReplicater.destroy();
	}

	@Override
	public ClientConfig select(Collection<ClientConfig> configs,RedisCommand command) {
		int hash=getHashCode(command);
		int slotIndex=Math.abs(hash)%slotManager.getSlotCount();
		
		String clientId=slotManager.getSlotClientId(slotIndex);
		if(StringUtils.isBlank(clientId)) {
			return null;
		}
		
		for(ClientConfig config:configs) {
			if(!config.getId().equals(clientId)) {
				continue;
			}
			
			return config;
		}
		
		return null;
	}
}